package com.gavin.plugin.service.impl;

import com.gavin.model.Request;
import com.gavin.model.Response;
import com.gavin.model.Result;
import com.gavin.model.RetCode;
import com.gavin.plugin.database.*;
import com.gavin.plugin.service.ITransBus;
import com.gavin.schema.dataservice.BigTable;
import com.gavin.schema.dataservice.TableRule;
import com.gavin.schema.service.DataGroup;
import com.gavin.schema.service.Database;
import com.gavin.schema.service.Parameter;
import com.gavin.schema.service.Plugin;
import com.gavin.utils.IOUtils;
import com.gavin.utils.StringUtils;
import com.gavin.utils.Utility;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.StringReader;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

/**
 * ------------------------------
 * 事务总线
 * ------------------------------
 * @author wdm（l311576@sina.com）  @date 2018年1月16日
 * @version 1.0
 */
public class TransBus implements ITransBus {
	private Logger logger = LoggerFactory.getLogger(TransBus.class);
	
	private HashMap<String, DataEngineList> hmDataGroup;
	private TransPlugin[] plugins;
	private HashMap<String, TransService> hmService;
	private HashMap<String, BigTable> hmTableRule;//表规则
	private ArrayList<TransService> alService;
	private IDataEngine dsEngine;
	private String strDataGroupId;
	private HashMap<String, String> hmParameterMap;
	
	//统计数据
	private AtomicBoolean abIsRunning = new AtomicBoolean();
	private AtomicLong alHandlingTransCount = new AtomicLong();
	
	public TransBus() {
		dsEngine = null;
		hmService = new HashMap<String, TransService>();
		alService = new ArrayList<TransService>();
		hmParameterMap = new HashMap<String, String>(10);
		hmTableRule = new HashMap<String, BigTable>();
	}
	
	/**
	 * 根据dbBusXML初始化对象
	 */
	public Result init(String dbBusXML, ClassLoader classLoader) {
		if (classLoader == null) {
			classLoader = this.getClass().getClassLoader();
		}

		// 将XML转换成对象
		com.gavin.schema.service.DBBus dBBus = null;
		try {
			dBBus = com.gavin.schema.service.DBBus.fromXML(dbBusXML);
		} catch (Exception e) {
			logger.error("Parse dbbus_s% , exception: ", e);
			return Result.valueOf(-1, "Initializing dbbus Failed: " + e.getLocalizedMessage());
		}
//		logger.info(dbBusXML);

		// 处理默认配置
		this.strDataGroupId = dBBus.getDataGroupId();

		// 初始化插件参数
		int nParameterCount = dBBus.getParameterCount();
		if (nParameterCount > 0) {
			if (logger.isInfoEnabled()) {
				logger.info("Loading transbus parameter ....................");
			}

			for (int i = 0; i < nParameterCount; i++) {
				Parameter para = dBBus.getParameter(i);
				this.hmParameterMap.put(para.getName(), para.getValue());
			}
		}

		// 初始化DataServiceEngine改名为DBTransEngine
		String dbTransEngine = dBBus.getDataServiceEngine();
		if (dbTransEngine == null || dbTransEngine.length() == 0) {
			this.dsEngine = new BigDataEngine();
		} else {
			try {
				logger.info("Loading DataServiceEngine " + dbTransEngine);
				this.dsEngine = (IDataEngine) Class.forName(dbTransEngine, true, classLoader).newInstance();
			} catch (Exception e) {
				logger.error("Load DataServiceEngine " + dbTransEngine + " failed, " + e.getMessage(), e);
				return Result.valueOf(-1,
						"Load DataServiceEngine " + dbTransEngine + " failed, " + e.getMessage());
			}
		}
		dsEngine.setTransBus(this);

		// 加载和初始化全局DataGroup
		this.hmDataGroup = new HashMap<String, DataEngineList>();
		DataGroup[] dgs = dBBus.getDataGroup();
		if (dgs != null && dgs.length > 0) {
			try {
				for (int i = 0; i < dgs.length; i++) {
					int nInstanceCount = dgs[i].getInstanceCount();
					if (nInstanceCount == 0) {// 单实例
						try {
							int nBigTableCount = dgs[i].getSubDatabaseCount();
							if (nBigTableCount <= 0) {// 单库
								if (logger.isInfoEnabled()) {
									logger.info("Initializing data group " + dgs[i].getId() + " ....................");
								}
								DataEngineList deList = dgs[i].init(classLoader);
								this.hmDataGroup.put(dgs[i].getId(), deList);
							} else {// 分库
								String strXML = dgs[i].toXML();
								for (int j = 0; j < nBigTableCount; j++) {
									DataGroup dg = DataGroup.fromXML(strXML);
									dg.setId(dg.getId().replaceFirst("#", "" + j));// Id
									for (int k = 0; k < dg.getDatabaseCount(); k++) {
										Database db = dg.getDatabase(k);
										db.setURI(db.getURI().replaceFirst("#", "" + j));// 库名
									}
									if (logger.isInfoEnabled()) {
										logger.info("Initializing data group " + dg.getId() + " ....................");
									}
									DataEngineList deList = dg.init(classLoader);
									this.hmDataGroup.put(dg.getId(), deList);
								}
							}
						} catch (Exception e) {
							logger.error( "Initializing data group " + dgs[i].getId() + " failed: " + e.getLocalizedMessage(), e);
							throw e;
						}
					} else {// 多实例
						try {
							int nBigTableCount = dgs[i].getSubDatabaseCount();
							if (nBigTableCount <= 0) {// 单库
								String strXML = dgs[i].toXML();
								for (int j = 0; j < nInstanceCount; j++) {
									DataGroup dg = DataGroup.fromXML(strXML);
									dg.setId(dg.getId().replaceFirst("#", "" + j));// Id
									for (int k = 0; k < dg.getDatabaseCount(); k++) {
										Database db = dg.getDatabase(k);
										db.setURI(db.getURI().replaceFirst("#", "" + j));// 域名
										db.setURI(db.getURI().replaceFirst("%", "" + j));// 端口号
									}
									if (logger.isInfoEnabled()) {
										logger.info("Initializing data group " + dg.getId() + " ....................");
									}
									DataEngineList deList = dg.init(classLoader);
									this.hmDataGroup.put(dg.getId(), deList);
								}
							} else {// 分库
								String strXML = dgs[i].toXML();
								for (int j = 0; j < nInstanceCount; j++) {
									for (int l = 0; l < nBigTableCount; l++) {
										DataGroup dg = DataGroup.fromXML(strXML);
										dg.setId(dg.getId().replaceFirst("#", "" + (j * nBigTableCount + l)));// Id
										for (int k = 0; k < dg.getDatabaseCount(); k++) {
											Database db = dg.getDatabase(k);
											db.setURI(db.getURI().replaceFirst("#", "" + j));// 域名
											db.setURI(db.getURI().replaceFirst("%", "" + j));// 端口号
											db.setURI(db.getURI().replaceFirst("#", "" + (j * nBigTableCount + l)));// 库名
										}
										if (logger.isInfoEnabled()) {
											logger.info(
													"Initializing data group " + dg.getId() + " ....................");
										}
										DataEngineList deList = dg.init(classLoader);
										this.hmDataGroup.put(dg.getId(), deList);
									}
								}
							}
						} catch (Exception e) {
							logger.error(
									"Initializing data group " + dgs[i].getId() + " failed: " + e.getLocalizedMessage(),
									e);
							throw e;
						}
					}
				}
			} catch (Exception e) {
				destroyDataGroup();
				this.hmDataGroup.clear();
				this.hmDataGroup = null;
				return Result.valueOf(-1, "Initializing transbus failed: " + e.getLocalizedMessage());
			}
		}

		// 加载分片规则
		Result ret = null;
		try {
			int tableRuleCount = dBBus.getTableRuleCount();
			for (int i = 0; i < tableRuleCount; i++) {
				com.gavin.schema.service.Sharding sharding = dBBus.getSharding(i);
				String shardingResource = sharding.getResource();
				if (logger.isInfoEnabled()) {
					logger.info("Loading sharding " + shardingResource + " ..............................");
				}
				String strXML = IOUtils.readTextResource(shardingResource, sharding.getEncoding(), classLoader);
				if (strXML == null) {
					throw new Exception("Sharding " + shardingResource + " is invalid");
				}
				
	    		TableRule tableRule = (TableRule)TableRule.unmarshal(new StringReader(strXML));
	    		BigTable[] bigTables= tableRule.getBigTable();
	    		for (int j = 0; j < bigTables.length; j++) {
	    			if(this.hmTableRule.get(bigTables[j].getName())!=null){
	    				throw new Exception("sharding rule [" + bigTables[j].getName() + "] is exist");
	    			}
	    			this.hmTableRule.put(bigTables[j].getName(),bigTables[j]);
	    			logger.info("Initializing sharding rule " + bigTables[j].toString());
				}
	    		
	    		
//	    		this.plugins[i] = new ServicePlugin();
//				// 初始化插件
//				this.plugins[i].setTransBus(this);
//				this.plugins[i].setBigMongoEngine(this.bmEngine);
//				this.plugins[i].setPublicDataGroup(hmDataGroup);
//				this.plugins[i].init(pluginResource, this, servicePluginDefine, classLoader);
			}
		} catch (Exception e) {
			logger.error("when parse plugin ,caught Exception: ", e);
			ret = Result.valueOf(-1, "Initializing transbus Failed: " + e.getLocalizedMessage());
			return ret;
		} finally {
			if (ret != null && ret.getCode() != 0) {
				this.destroy();
			}
		}		
		
		
		// 加载插件对象
		ret = null;
		try {
			int nPluginCount = dBBus.getPluginCount();
			this.plugins = new TransPlugin[nPluginCount];
			for (int i = 0; i < nPluginCount; i++) {
				Plugin plugin = dBBus.getPlugin(i);
				String pluginResource = plugin.getResource();
				if (logger.isInfoEnabled()) {
					logger.info("Loading plugin " + pluginResource + " ..............................");
				}
				String strXML = IOUtils.readTextResource(pluginResource, plugin.getEncoding(), classLoader);
				if (strXML == null) {
					throw new Exception("Plugin " + pluginResource + " is invalid");
				}
				com.gavin.schema.service.ServicePlugin servicePluginDefine = com.gavin.schema.service.ServicePlugin
						.fromXML(strXML);
				this.plugins[i] = new TransPlugin();

				// 初始化插件
//				this.plugins[i].setTransBus(this);
				this.plugins[i].setPublicDataGroup(hmDataGroup);
				this.plugins[i].init(pluginResource, this, servicePluginDefine, classLoader);
			}
		} catch (Exception e) {
			logger.error("Parse plugin Exception: ", e);
			ret = Result.valueOf(-1, "Initializing transbus Failed: " + e.getLocalizedMessage());
			return ret;
		} finally {
			if (ret != null && ret.getCode() != 0) {
				this.destroy();
			}
		}
		
		return Result.OK;
	}

	/**
	 * 启动Business服务
	 * 
	 * @return
	 */
	public Result start() {
		for (int i = 0; i < this.alService.size(); i++) {
			alService.get(i).start();
		}
		abIsRunning.set(true);

		return Result.OK;
	}

	/**
	 * 停止Business服务
	 *
	 */
	public void stop() {
		while (true) {
			long lHandlingTransCount = this.alHandlingTransCount.get();
			int nWaitingTaskCount = 0;
			if (lHandlingTransCount + nWaitingTaskCount > 0) {
				logger.info("Try to stop, now waiting for handling trans/task done: " + lHandlingTransCount + '/' + nWaitingTaskCount + ", please stop request-client first");
				Utility.sleep(1000);
				continue;
			}
			break;
		}
		abIsRunning.set(false);

		// Stop Plugin
		for (int i = 0; i < this.alService.size(); i++) {
			alService.get(i).stop();
		}
		this.destroy();
	}
	
	public void destroy() {
		this.alService.clear();
		this.plugins = null;
		this.destroyDataGroup();
		this.hmDataGroup.clear();
		this.hmParameterMap.clear();
		this.hmService.clear();
	}

	private void destroyDataGroup() {
		for (DataEngineList cl : this.hmDataGroup.values()) {
			if (cl == null) {
				continue;
			}
			int nCount = cl.size();
			for (int i = 0; i < nCount; i++) {
				DataEngine dataEngine = cl.get(i);
				if (dataEngine == null) {
					continue;
				}
				dataEngine.close();
			}
		}
	}

	public TransService addService(TransService transService) {
		String serviceName = transService.getServiceName().toLowerCase();
		if (this.hmService.containsKey(serviceName)) {
			return null;
		}
		this.hmService.put(serviceName, transService);
		alService.add(transService);
		return transService;
	}
	
	public IDataEngine getDataServiceEngine() {
		return this.dsEngine;
	}
	public String getParameterValue(String strParameterName) {
		return hmParameterMap.get(strParameterName);
	}
	public boolean isRunning() {
		return this.abIsRunning.get();
	}
	public long getHandlingTransCount() {
		return this.alHandlingTransCount.get();
	}
	public void setDataGroupId(String strGroupId) {
		this.strDataGroupId = strGroupId;
	}	
	public String getDataGroupId() {
		return this.strDataGroupId;
	}

	public BigTable getTableRule(String tableName) {
		return this.hmTableRule.get(tableName);
	}
	
	public String getParameter(String strName, String strDefaultValue) {
		String strValue = hmParameterMap.get(strName);
		if (strValue == null) {
			return strDefaultValue;
		}
		return strValue;
	}
	public String getParameter(String strName) {
		return hmParameterMap.get(strName);
	}

	/**
	 * 调用事务处理的方法
	 */
	public Response execute(Request request) {
		String serviceName = request.getServiceName();
		String transName = request.getTransName();
		if(StringUtils.isBlank(serviceName) || StringUtils.isBlank(transName)){
			logger.error("Invalid serviceTrans request");
			return new Response(RetCode.REQUEST_FAIL.getState(),"无效的请求");
		}
		
		try {
			// 正常处理
			TransService transService = this.hmService.get(serviceName.toLowerCase());
			if (transService != null) {
				this.alHandlingTransCount.incrementAndGet();
				return transService.execute(request);
			}else{
				logger.error("service "+serviceName+"  not support");
				return new Response(RetCode.REQUEST_FAIL.getState(),"不支持的请求");
			}
		} catch (Exception e) {
			logger.error("handleTrans " + serviceName + '.' + transName + " failed:" + e.getMessage(), e);
			this.alHandlingTransCount.decrementAndGet();
			return new Response(RetCode.REQUEST_FAIL.getState(), "请求执行异常");
		}
	}

	@Override
	public Response execute(IAtom iAtom) {
		Request request=iAtom.getRequest();
		String serviceName = request.getServiceName();
		String transName = request.getTransName();
		if(StringUtils.isBlank(serviceName) || StringUtils.isBlank(transName)){
			logger.error("Invalid serviceTrans request");
			return new Response(RetCode.REQUEST_FAIL.getState(),"无效的请求");
		}

		try {
			// 正常处理
			TransService transService = this.hmService.get(serviceName.toLowerCase());
			if (transService != null) {
				this.alHandlingTransCount.incrementAndGet();
				return transService.execute(iAtom);
			}else{
				logger.error("service "+serviceName+"  not support");
				return new Response(RetCode.REQUEST_FAIL.getState(),"不支持的请求");
			}
		} catch (Exception e) {
			logger.error("handleTrans " + serviceName + '.' + transName + " failed:" + e.getMessage(), e);
			this.alHandlingTransCount.decrementAndGet();
			return new Response(RetCode.REQUEST_FAIL.getState(), "请求执行异常");
		}
	}
}
